package com.navinfo.tripanalysis.common.arithmetic.convert;

import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.protocol.java.EventDataPb;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;

/**
 * 事件数据与Parquet格式转换
 *
 * @author 沈东生
 */
public class EventDataParquetConvert {

    /**
     * 创建Schema
     *
     * @return
     */
    public static StructType createSchema() {
        List<StructField> list = new ArrayList<>();
        /*--start_schema*/
        list.add(new StructField("uid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("tid", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("stype", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_gps_time", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("start_latitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_longitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_height", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_speed", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("start_gps_direction", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_gps_time", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("stop_latitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_longitude", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_height", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_speed", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("stop_gps_direction", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("duration", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("event", DataTypes.StringType, true, Metadata.empty()));
        list.add(new StructField("extend_field01", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field02", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field03", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field04", DataTypes.IntegerType, true, Metadata.empty()));
        list.add(new StructField("extend_field05", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field06", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field07", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field08", DataTypes.DoubleType, true, Metadata.empty()));
        list.add(new StructField("extend_field09", DataTypes.LongType, true, Metadata.empty()));
        list.add(new StructField("extend_field10", DataTypes.LongType, true, Metadata.empty()));
        /*--end_schema*/

        return new StructType(list.toArray(new StructField[list.size()]));
    }

    /**
     * 创建一行数据
     *
     * @param m
     * @param schema
     * @return
     */
    public static Row createRow(OuterEventData m, StructType schema) {
        List<Object> list = new ArrayList<>();
        /*--start_row*/
        list.add(m.getUid());
        list.add(m.getTid());
        list.add(m.getStype());
        list.add(m.getStartGpsTime());
        list.add(m.getStartLatitude());
        list.add(m.getStartLongitude());
        list.add(m.getStartHeight());
        list.add(m.getStartSpeed());
        list.add(m.getStartGpsDirection());
        list.add(m.getStopGpsTime());
        list.add(m.getStopLatitude());
        list.add(m.getStopLongitude());
        list.add(m.getStopHeight());
        list.add(m.getStopSpeed());
        list.add(m.getStopGpsDirection());
        list.add(m.getDuration());
        list.add(StatisticDataParquetConvert.toJsonStr(m.getEvent()));
        list.add(m.getExtendField01());
        list.add(m.getExtendField02());
        list.add(m.getExtendField03());
        list.add(m.getExtendField04());
        list.add(m.getExtendField05());
        list.add(m.getExtendField06());
        list.add(m.getExtendField07());
        list.add(m.getExtendField08());
        list.add(m.getExtendField09());
        list.add(m.getExtendField10());
        /*--end_row*/

        return new GenericRowWithSchema(list.toArray(), schema);
    }

    /**
     * Row转为Pojo对象
     *
     * @param row
     * @return
     */
    public static OuterEventData fromRow(Row row) {
        OuterEventData m = new OuterEventData();
        /*--start_deserialization*/
        m.setUid(StatisticDataParquetConvert.getLong(row,"uid"));
        m.setTid(StatisticDataParquetConvert.getLong(row,"tid"));
        m.setStype(StatisticDataParquetConvert.getInteger(row,"stype"));
        m.setStartGpsTime(StatisticDataParquetConvert.getLong(row,"start_gps_time"));
        m.setStartLatitude(StatisticDataParquetConvert.getInteger(row,"start_latitude"));
        m.setStartLongitude(StatisticDataParquetConvert.getInteger(row,"start_longitude"));
        m.setStartHeight(StatisticDataParquetConvert.getInteger(row,"start_height"));
        m.setStartSpeed(StatisticDataParquetConvert.getInteger(row,"start_speed"));
        m.setStartGpsDirection(StatisticDataParquetConvert.getInteger(row,"start_gps_direction"));
        m.setStopGpsTime(StatisticDataParquetConvert.getLong(row,"stop_gps_time"));
        m.setStopLatitude(StatisticDataParquetConvert.getInteger(row,"stop_latitude"));
        m.setStopLongitude(StatisticDataParquetConvert.getInteger(row,"stop_longitude"));
        m.setStopHeight(StatisticDataParquetConvert.getInteger(row,"stop_height"));
        m.setStopSpeed(StatisticDataParquetConvert.getInteger(row,"stop_speed"));
        m.setStopGpsDirection(StatisticDataParquetConvert.getInteger(row,"stop_gps_direction"));
        m.setDuration(StatisticDataParquetConvert.getLong(row,"duration"));
        //ignore set setEvent 
        m.setExtendField01(StatisticDataParquetConvert.getInteger(row,"extend_field01"));
        m.setExtendField02(StatisticDataParquetConvert.getInteger(row,"extend_field02"));
        m.setExtendField03(StatisticDataParquetConvert.getInteger(row,"extend_field03"));
        m.setExtendField04(StatisticDataParquetConvert.getInteger(row,"extend_field04"));
        m.setExtendField05(StatisticDataParquetConvert.getDouble(row,"extend_field05"));
        m.setExtendField06(StatisticDataParquetConvert.getDouble(row,"extend_field06"));
        m.setExtendField07(StatisticDataParquetConvert.getDouble(row,"extend_field07"));
        m.setExtendField08(StatisticDataParquetConvert.getDouble(row,"extend_field08"));
        m.setExtendField09(StatisticDataParquetConvert.getLong(row,"extend_field09"));
        m.setExtendField10(StatisticDataParquetConvert.getLong(row,"extend_field10"));
        /*--end_deserialization*/
        String eventName= StatisticDataParquetConvert.getString(row,"event");
        m.setEvent(EventDataPb.EventType.valueOf(eventName));
        m.setDay(StatisticDataParquetConvert.getString(row,"day"));
        return m;
    }
}
